package com.lagou.rpc.consumer.zookeeper;

import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static org.apache.curator.framework.recipes.cache.CuratorCacheListener.builder;

/**
 * @author mcl
 * @date 2021/7/29 1:42 下午
 * @description
 */
@Slf4j
@Component
public class ZkNode {

    @Autowired
    private RpcClientProxy rpcClientProxy;

    private CuratorFramework client;
    /**
     * 注册监听
     * TreeCache: 可以将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，
     * 呈现树形目录的监听，可以设置监听深度，最大监听深度为 int 类型的最大值。
     */
    public void watchNode() throws Exception {
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(exponentialBackoffRetry)
                .namespace("lg-rpc")  // 独立的命名空间 /base
                .build();

        client.start();

        String temp_path = "/server";

//        CountDownLatch watch = new CountDownLatch(1);
//        TreeCache treeCache = new TreeCache(client, temp_path);
//        treeCache.getListenable().addListener(new TreeCacheListener() {
//            @Override
//            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
//                ChildData eventData = event.getData();
//                switch (event.getType()) {
//                    case NODE_ADDED:
//                        log.info(temp_path + "节点添加" + eventData.getPath() + "\t添加数据为：" + new String(eventData.getData()));
//                        String addData = new String(eventData.getData());
//                        if (addData.contains(":")) {
//                            RpcClientProxy.addClient(addData.split(":")[0], Integer.parseInt(addData.split(":")[1]));
//                        }
//                        break;
//                    case NODE_UPDATED:
//                        log.info(eventData.getPath() + "节点数据更新\t更新数据为：" + new String(eventData.getData()) + "\t版本为：" + eventData.getStat().getVersion());
//                        break;
//                    case NODE_REMOVED:
//                        log.info(eventData.getPath() + "节点被删除");
//                        String delData = new String(eventData.getData());
//                        if (delData.contains(":")) {
//                            RpcClientProxy.removeClient(delData.split(":")[0], Integer.parseInt(delData.split(":")[1]));
//                        }
//                        break;
//                    default:
//                        break;
//                }
//            }
//        });
//        treeCache.start();
//        watch.await();  //如果不执行 watch.countDown()，进程会一致阻塞在 watch.await()

        List<String> strings = client.getChildren().forPath(temp_path);
        strings.forEach(s -> {
            try {
                byte[] bytes = client.getData().forPath(temp_path + "/" + s);
                log.info(temp_path + "/" + s + "节点添加" + s + "\t添加数据为：" + new String(bytes));
                String addData = new String(bytes);
                if (addData.contains(":")) {
                    rpcClientProxy.addClient(addData.split(":")[0],
                            Integer.parseInt(addData.split(":")[1]),
                            temp_path + "/" + s,
                            Long.parseLong(addData.split(":")[2]),
                            Long.parseLong(addData.split(":")[3]));                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        CuratorCache cache = CuratorCache.build(client, temp_path);
        CountDownLatch initializedLatch = new CountDownLatch(1);
        AtomicInteger eventCount = new AtomicInteger(0);
        CuratorCacheListener listener = new CuratorCacheListener()
        {
            @Override
            public void event(CuratorCacheListener.Type type, ChildData oldData, ChildData data)
            {
                eventCount.incrementAndGet();

                switch (type) {
                    case NODE_CREATED:
                        log.info(temp_path + "节点添加" + data.getPath() + "\t添加数据为：" + new String(data.getData()));
                        String addData = new String(data.getData());
                        if (addData.contains(":")) {
//                            rpcClientProxy.addClient(addData.split(":")[0], Integer.parseInt(addData.split(":")[1]));
                            rpcClientProxy.addClient(addData.split(":")[0],
                                    Integer.parseInt(addData.split(":")[1]),
                                    data.getPath(),
                                    Long.parseLong(addData.split(":")[2]),
                                    Long.parseLong(addData.split(":")[3]));
                        }
                        break;
                    case NODE_CHANGED:
                        log.info(data.getPath() + "节点数据更新\t更新数据为：" + new String(data.getData()) + "\t版本为：" + data.getStat().getVersion());
                        String updateData = new String(data.getData());
                        rpcClientProxy.updateClient(updateData.split(":")[0],
                                Integer.parseInt(updateData.split(":")[1]),
                                Long.parseLong(updateData.split(":")[2]),
                                Long.parseLong(updateData.split(":")[3]));
                        break;
                    case NODE_DELETED:
                        log.info(oldData.getPath() + "节点被删除");
                        String delData = new String(oldData.getData());
                        if (delData.contains(":")) {
                            rpcClientProxy.removeClient(delData.split(":")[0], Integer.parseInt(delData.split(":")[1]));
                        }
                        break;
                    default:
                        break;
                }
            }

            @Override
            public void initialized()
            {
                initializedLatch.countDown();
            }
        };
        cache.listenable().addListener(builder().forAll(listener).afterInitialized().build());
        cache.start();
    }

    public void uploadZkNodeState(RpcClient rpcClient) {
        //        ip : port : lastResponseTime : lastResponseSpendTime
        if (client == null || rpcClient == null) {
            return;
        }
        String serverInfo = rpcClient.getIp() + ":" + rpcClient.getPort() + ":" + rpcClient.getLastResponse();
        BigDecimal lastTime = new BigDecimal(rpcClient.getLastResponseTime());
        BigDecimal currentTime = new BigDecimal(System.currentTimeMillis());
        BigDecimal subtract = currentTime.subtract(lastTime).divide(new BigDecimal(1000), RoundingMode.UP).subtract(new BigDecimal(5));
        try {
            if (subtract.intValue() > 0) {
                client.delete().forPath(rpcClient.getZkPath());
            } else {
                client.setData().forPath(rpcClient.getZkPath(), serverInfo.getBytes());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
